package com.ytktt.study.week12.task4.kafka.consumer;

import com.ytktt.study.week12.task4.Order;
import com.ytktt.study.week12.task4.kafka.constant.KafkaConstant;
import com.ytktt.study.week12.task4.redis.constant.RedisConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * kafka 消息序列消费者者拦截器
 * 幂等参考 <a href="https://github.com/Jaskey/RocketMQDedupListener/blob/master/src/main/java/com/github/jaskey/rocketmq/strategy/DedupConsumeStrategy.java">RocketMQ的幂等示例</a>
 *
 * @author ytk
 */
@Slf4j
public class KafkaSeqConsumerInterceptor implements ConsumerInterceptor<String, Order> {

    private RedisTemplate<String, Object> redisTemplate;
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public ConsumerRecords<String, Order> onConsume(ConsumerRecords<String, Order> records) {
        for (ConsumerRecord<String, Order> record : records) {
            Order order = record.value();
            this.checkSeq(record);
            String orderId = String.valueOf(order.getId());
            this.redisTemplate.opsForZSet().add(String.format(RedisConstant.ORDER_CONSUMER_PREPARE_CONSUME_SET, record.partition()),
                    orderId, record.offset());
            this.redisTemplate.opsForHash().put(RedisConstant.ORDER_INFO, orderId, order);
        }
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
            String key = String.format(RedisConstant.ORDER_CONSUMER_PREPARE_CONSUME_SET, entry.getKey().partition());
            long offset = entry.getValue().offset();
            Set<Object> orderIds = this.redisTemplate.opsForZSet().rangeByScore(key, offset, offset);
            if (CollectionUtils.isNotEmpty(orderIds)) {
                for (Object id : orderIds) {
                    this.redisTemplate.opsForZSet().remove(key, id);
                    this.redisTemplate.opsForHash().delete(RedisConstant.ORDER_INFO, id);
                }
            }
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        this.redisTemplate = (RedisTemplate<String, Object>) configs.get(KafkaConstant.REDIS_TEMPLATE_CONFIG);
        this.stringRedisTemplate = (StringRedisTemplate) configs.get(KafkaConstant.REDIS_STRING_TEMPLATE_CONFIG);
    }

    private void checkSeq(ConsumerRecord<String, Order> record) {
        Order order = record.value();
        Long seq = order.getMessageInfo().getSeq();
        String seqKey = String.format(RedisConstant.ORDER_CONSUMER_MESSAGE_SEQ, order.getMessageInfo().getDate(), record.partition());
        String o = this.stringRedisTemplate.opsForValue().get(seqKey);
        if (o != null) {
            Long curSeq = Long.valueOf(o);
            if (seq - curSeq != 1) {
                log.warn("lost record: {}", record);
            } else {
                // stringTemplate解决取出序列号后变为Integer的问题
                this.stringRedisTemplate.opsForValue().set(seqKey, String.valueOf(seq));
                this.stringRedisTemplate.expire(seqKey, 2, TimeUnit.DAYS);
            }
        } else {
            this.stringRedisTemplate.opsForValue().set(seqKey, String.valueOf(seq));
            this.stringRedisTemplate.expire(seqKey, 2, TimeUnit.DAYS);
        }
    }
}
